import org.apache.spark.util.AccumulatorV2

import scala.collection.mutable

/**
 * 自定义累加器 AccumulatorV2[IN, OUT]
 */
class SessionAccumulator extends AccumulatorV2[String, mutable.HashMap[String, Int]] {
  // 创建一个自定义累加器的结构，自定义累加器就是对countmap结构的维护
  val countMap = new mutable.HashMap[String, Int]()

  // 判断累加器是否为空
  override def isZero: Boolean = {
    countMap.isEmpty
  }

  // 当前的累加器去copy一个与当前累加器一模一样结构的累加器
  override def copy(): AccumulatorV2[String, mutable.HashMap[String, Int]] = {
    val acc = new SessionAccumulator
    acc.countMap ++= this.countMap // 将两个map拼接到一起，++= 为合并map
    acc
  }


  override def reset(): Unit = {
    countMap.clear()
  }

  override def add(v: String): Unit = {
    // 判断传入的v是否在countmap当中
    if (!this.countMap.contains(v)) {
      // 不在，则给当前的v创建一个k-v值
      this.countMap += (v -> 0)
    }
    // 计数+1
//    this.countMap.update(v, countMap(v) + 1)
    this.countMap(v) += 1
  }

  // merge ,两个累加器进行合并
  override def merge(other: AccumulatorV2[String, mutable.HashMap[String, Int]]): Unit = {
    // 要匹配的值  match{}
    other match {
      // (0 /: (1 to 100))(_+_)
      // (0 /: (1 to 100)){case (int1, !) => int1 + int2}
      // (1 /: 100).foldLeft(0)
      // (this.countMap /: acc.countMap)
      //map.foldLeft(初始值){操作函数}
      //this.countMap作为初始值对acc.countmap做操作，this.countMap和acc.countMap都是map
      case acc: SessionAccumulator => acc.countMap.foldLeft(this.countMap) {
            //1、map相当于this.countMap,(k,v)相当于acc.countMap
            //2、map += （k+v） 合并map
        case (map, (k, v)) => map += (k -> (map.getOrElse(k, 0) + v))
      }
    }

  }

  override def value: mutable.HashMap[String, Int] = {
    this.countMap
  }
}
